package test

import cn.hutool.core.util.RandomUtil
import cn.hutool.log.Log
import com.gitee.ixtf.guice.Jguice
import com.gitee.ixtf.vertx.guice.VertxGuice
import com.gitee.ixtf.vertx.rabbitmq.guice.RabbitMQGuice
import com.gitee.ixtf.vertx.rabbitmq.queueMessage
import com.gitee.ixtf.vertx.rabbitmq.send
import com.gitee.ixtf.vertx.rabbitmq.verticle.RabbitMQConsumerVerticle
import io.vertx.core.ThreadingModel
import io.vertx.core.Vertx
import io.vertx.core.eventbus.EventBus
import io.vertx.core.eventbus.Message
import io.vertx.kotlin.core.deploymentOptionsOf
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.rabbitmq.RabbitMQMessage
import io.vertx.rabbitmq.RabbitMQOptions
import java.util.concurrent.TimeUnit
import kotlin.system.exitProcess
import kotlin.text.Charsets.UTF_8
import kotlin.time.Duration.Companion.seconds
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collect

private val vertx: Vertx by Jguice
private val eventBus: EventBus by Jguice
private val channel = Channel<Any>()

suspend fun main() {
  runCatching {
        Jguice.init(
            VertxGuice(Vertx.vertx()),
            object : RabbitMQGuice() {
              override fun prepare(options: RabbitMQOptions) {
                options.host = "git.medipath.com.cn"
                options.user = "admin"
                options.password = "medipath888"
              }
            })
        Runtime.getRuntime().addShutdownHook(Thread { runBlocking { vertx.close().coAwait() } })
        vertx
            .deployVerticle(
                TestVerticle(),
                deploymentOptionsOf(
                    threadingModel = ThreadingModel.WORKER,
                    maxWorkerExecuteTime = 1,
                    maxWorkerExecuteTimeUnit = TimeUnit.HOURS))
            .coAwait()
      }
      .onFailure { exitProcess(1) }

  (1..10).map { queueMessage("test", "$it") }.send()

  delay(3.seconds)
  eventBus.request<Boolean>("test", "publish").coAwait()
}

private class TestVerticle : RabbitMQConsumerVerticle() {
  override val queue = "test"
  override val error = "test:Error"
  override val prefetchCount = 5
  override val onSuccessAck = false

  override suspend fun start() {
    eventBus.coConsumer("test") { handleTest(it) }
    super.start()
  }

  override suspend fun handle(delivery: RabbitMQMessage) {
    log.info("handle: ${delivery.body().toString(UTF_8)}")
    AlgTask(delivery) { basicAck() }.invoke()
  }

  private suspend fun handleTest(reply: Message<Any>) {
    delay(5.seconds)
    reply.reply(true)
    log.info("coConsumer: ${reply.address()}")
  }
}

data class AlgTask(
    private val delivery: RabbitMQMessage,
    private val onSuccess: suspend RabbitMQMessage.() -> Unit
) {
  private val log by lazy { Log.get(this::class.java) }
  private val testFlow = channelFlow {
    (1..10).asFlow().collect {
      launch {
        delay(RandomUtil.randomInt(3, 5).seconds)
        send(it)
        log.info("send")
      }
    }
  }

  suspend fun invoke() {
    // testDelay()

    testChannel()

    log.info("delay: ${delivery.body().toString(UTF_8)}")
    onSuccess(delivery)
    log.info("onSuccess: ${delivery.body().toString(UTF_8)}")
  }

  private suspend fun testChannel() {
    testFlow.collect()
  }

  private suspend fun testDelay() {
    withContext(Dispatchers.IO) {
      delay(RandomUtil.randomInt(3, 5).seconds)
      log.info("delay: ${delivery.body().toString(UTF_8)}")
    }
  }
}
